package com.cctc.rds.scheduler.client.worker.impl;

import com.cctc.rds.nrpc.core.util.IdUtils;
import com.cctc.rds.nrpc.core.util.JsonUtils;
import com.cctc.rds.scheduler.client.worker.TaskWorker;
import com.cctc.rds.scheduler.client.worker.TaskWorkerCallback;
import com.cctc.rds.scheduler.core.task.Task;
import com.cctc.rds.scheduler.core.task.TaskFactory;
import com.cctc.rds.scheduler.core.task.TaskLog;
import com.cctc.rds.scheduler.core.task.impl.TaskAdapter;

import java.io.IOException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/**
 * 任务（自动）执行 [接口] 实现类。
 * 描述信息：任务调度器客户端自动从 Leader 节点的执行容器中拉取任务并执行，执行完成后向 Leader 节点更新任务执行结果（任务状态：success、fail），并将
 * 更新后的任务实例日志（任务状态：success、fail）同步至所有 Follower 节点；
 * --> 如果任务实例日志状态为 success（执行成功），则结束执行。
 * --> 如果任务实例日志状态为 fail（执行失败），则设置重新调度时间为当前时间，并计算下一次延迟执行时间，更新任务状态为 pending。
 */
public class DefaultTaskWorker implements TaskWorker {
    private final String app;
    private ExecutorService workerGroup;

    public DefaultTaskWorker(String app, int threads) {
        this.app = app;
        this.workerGroup = new ThreadPoolExecutor(
                threads, threads,
                60, TimeUnit.SECONDS,
                new LinkedBlockingDeque<>(),
                r -> new Thread(r, "tast-run-worker-" + IdUtils.newId()),
                new ThreadPoolExecutor.DiscardPolicy()
        );
    }

    @Override
    public void run(TaskLog taskLog, TaskWorkerCallback callback) {
        Task<?> task = TaskFactory.getTask(app, taskLog.getTaskName());
        Task<Object> action = new TaskAdapter<Object>((Task<Object>) task) {
            @Override
            public Object getTaskParam() {
                return JsonUtils.fromJson(taskLog.getTaskParam(), task.getTaskParamType());
            }
        };
        workerGroup.execute(() -> {
            boolean success = false;
            Throwable e = null;
            try {
                int curExecPeriodIndex = taskLog.getExecPriodIndex();
                success = action.run(curExecPeriodIndex, action.getTaskParam());
            } catch (Throwable throwable) {
                e = throwable;
            }
            callback.onFinish(taskLog.getTaskId(), success, e);
        });
    }

    @Override
    public void close() throws IOException {
        workerGroup.shutdownNow();
    }
}
